package job.qi.dftf;

import java.io.IOException;
import java.util.List;

import mapred.example.ParseReader;
import mapred.example.model.Link;
import mapred.example.model.Page;
import mapred.example.model.Paragraph;
import mapred.util.TermCounter;

import org.apache.hadoop.io.Text;

public class Qi_DFTF_Ready_Mapper extends ParseReader {

    @Override
    public void map(Page page) throws IOException {
        String KEY = page.getUrl().getNormalized();

        {
            TermCounter url = listToTermCounter(page.getUrl().getText());
            if (!url.isEmpty()) {
                this.output(new Text(KEY), new Text("URL\t" + url));
            }
        }
        {
            TermCounter title = listToTermCounter(page.getTitle());
            if (!title.isEmpty()) {
                this.output(new Text(KEY), new Text("TITLE\t" + title));
            }
        }
        {
            TermCounter body = new TermCounter();
            for (Paragraph p : page.getBody()) {
                TermCounter sub = listToTermCounter(p.getText());
                body.add(sub);
            }
            if (!body.isEmpty()) {
                this.output(new Text(KEY), new Text("BODY\t" + body));
            }
        }
        {
            for (Link l : page.getOutLinks()) {
                TermCounter anchor = listToTermCounter(l.getText());
                if (!anchor.isEmpty()) {
                    this.output(new Text(l.getUrl()), new Text("ANCHOR\t"
                            + anchor));
                }
            }
        }
    }

    private TermCounter listToTermCounter(List<String> list) {
        TermCounter ret = new TermCounter();
        for (String s : list) {
            ret.add(s);
        }
        return ret;
    }

    @Override
    public void setUp() {
    }

    @Override
    public void tearDown() throws IOException {
    }

}
